package com.fyy.kafuka.utils;

import org.apache.kafka.common.internals.Topic;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;

/**
 * @Author fyy
 * @Description kafuka消息发送
 * @Date 2020/10/9 14:29
 */
@Component
public class MsgLafukaSendUtils {

    private static Logger logger = LoggerFactory.getLogger(MsgLafukaSendUtils.class);
    @Autowired
    private KafkaTemplate kafkaTemplate;

    /**
     * @Author fyy
     * @Description 发送消息
     * @param context 消息内容
     * @Date 2020/10/9 14:30
     */
    public void sendMsg(String context){
        ListenableFuture futures = kafkaTemplate.send("aaa",Topic.LEGAL_CHARS,context);
        futures.addCallback(
                o -> logger.info("发送消息成功：{}",context),
                throwable -> logger.error("发送消息失败：{}",throwable.getMessage())
        );
    }

}
